package com.lengxf.flink.window;

import com.lengxf.flink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * 窗口类型的demo
 *
 * @author Lengxf
 */
public class WindowApiDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        DataGeneratorSource<WaterSensor> source = new DataGeneratorSource<>(
                (GeneratorFunction<Long, WaterSensor>) aLong -> new WaterSensor("id1", Math.toIntExact(aLong), aLong),
                100,
                RateLimiterStrategy.perSecond(1),
                Types.POJO(WaterSensor.class)
        );

        KeyedStream<WaterSensor, String> keyedStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "data-generator")
                .keyBy(e -> e.id);
        //没有keyBy的窗口  窗口内的所有数据都会进入同一个子任务  并行度强制为1
        //窗口算子无法设置并行度
        //基于时间的滚动窗口  窗口窗口为3S
        WindowedStream<WaterSensor, String, TimeWindow> window = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(3)));
        //滑动窗口 窗口长度10s 滑动步长2s
        //keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(2)));
        //会话窗口
        //keyedStream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)));

        //基于计数的  每5条数据就会生成一个窗口
        //keyedStream.countWindow(5);
        //基于技术的滑动窗口
        //keyedStream.countWindow(5, 2);
        //全局窗口 需要自定义触发器   需要自定义的时候 才会用
        //keyedStream.window(GlobalWindows.create());

        //增量聚合  来一条数据 就计算一条数据  窗口触发的时候 输出计算结果
        window.reduce((ReduceFunction<WaterSensor>) (value1, value2) -> {
            System.out.println("触发v1 = " + value1);
            System.out.println("触发v2 = " + value2);
            return new WaterSensor(value1.getId(), value2.getVc() + value1.getVc(), value1.getTs());
        }).print();
        env.execute();
    }
}
